package com.crpc.core.proxy.handler;

import cn.hutool.core.util.ObjectUtil;
import com.crpc.common.annotation.TryTimes;
import com.crpc.common.crpcEnum.CrpcExceptionEnums;
import com.crpc.common.crpcEnum.RequestType;
import com.crpc.common.exception.CrpcException;
import com.crpc.core.CrpcBootStrap;
import com.crpc.core.NettyBootstrapInitializer;
import com.crpc.core.compress.CompressorFactory;
import com.crpc.core.discovery.Registry;
import com.crpc.core.serialize.SerializerFactory;
import com.crpc.core.transport.message.CrpcRequest;
import com.crpc.core.transport.message.RequestPayLoad;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.util.Date;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

import static com.crpc.core.CrpcBootStrap.CHANNEL_MAP;
import static com.crpc.core.CrpcBootStrap.CRPC_REQUEST_THREAD_LOCAL;

/**
 * @author CLOUD
 * @version 1.0
 * @description:
 * @date 2023/9/10 13:47
 */
@Data
@Slf4j
public class RpcConsumerInvocationHandler implements InvocationHandler {
    private Class<?> interfaceProvider;
    private Object ref;
    private Registry registry;

    public RpcConsumerInvocationHandler(Registry registry, Class<?> interfaceProvider) {
        this.registry = registry;
        this.interfaceProvider = interfaceProvider;
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        //反射获取 method  的注解 并且筛选出 TryTimes 注解的方法
        int interValTime = 0;
        int tryTimes = 0;
        TryTimes tryTimesAnnotation = method.getAnnotation(TryTimes.class);
        if (!ObjectUtil.isNull(tryTimesAnnotation)) {
            interValTime = tryTimesAnnotation.intervalTime();
            tryTimes = tryTimesAnnotation.tryTimes();
        }
        while (true) {
            try {
                /*
                 * ======================封装报文=================
                 */
                RequestPayLoad requestPayLoad = RequestPayLoad.builder()
                        .interfaceName(interfaceProvider.getName())
                        .methodName(method.getName())
                        .parametersType(method.getParameterTypes())
                        .parametersValue(args)
                        .returnType(method.getReturnType())
                        .build();

                CrpcRequest crpcRequest = CrpcRequest.builder()
                        .requestId(CrpcBootStrap.getInstance().getConfiguration().getIdGenerator().getId())
                        .compressType(CompressorFactory.getCompressor(CrpcBootStrap.getInstance().getConfiguration().getCompressorType()).getCode())
                        .requestType(RequestType.REQUEST.getId())
                        .serializeType(SerializerFactory.getSerializer(CrpcBootStrap.getInstance().getConfiguration().getSerializerType()).getCode())
                        .timeStamp(new Date().getTime())
                        .requestPayLoad(requestPayLoad)
                        .build();
                CRPC_REQUEST_THREAD_LOCAL.set(crpcRequest);
                //使用负载均衡器去服务列表寻找一个服务
                InetSocketAddress inetSocketAddress = CrpcBootStrap.getInstance().getConfiguration().getLoadBalance().selectServiceAddress(interfaceProvider.getName());
                //使用服务器连接netty,发送调用, 服务的名字,方法的名字,参数列表
                log.info("服务调用方发现，地址：{},端口：{}", inetSocketAddress.getAddress(), inetSocketAddress.getPort());
                //获取一个可用通道
                Channel channel = createAvailableChannel(inetSocketAddress);
                /*
                 *   ======================异步策略=================
                 */
                //写出报文
                CompletableFuture<Object> future = new CompletableFuture<>();
                CrpcBootStrap.PENDING_REQUEST.put(crpcRequest.getRequestId(), future);
                //写出一个请求,这个歌请求的实例会进入pipeline执行出栈一系列操作
                //第一个出栈程序一定是个crpcRequest--->后续转换为二进制报文
                channel.writeAndFlush(crpcRequest)
                        .addListener((ChannelFutureListener) promise -> {

                            if (!promise.isSuccess()) {
                                future.completeExceptionally(promise.cause());
                            }
                        });
                return future.get(3, TimeUnit.SECONDS);
            } catch (Exception e) {
                log.error("调用失败,剩余尝试次数：{},错误信息：{}", tryTimes, e.getMessage(), e);
                tryTimes--;
                Thread.sleep(interValTime);
                if (tryTimes < 0) {
                    log.error("对方法：{}调用失败", method.getName());
                    break;
                }
            }
        }
        throw new CrpcException(CrpcExceptionEnums.REQUEST_TIMEOUT.getCode(), "调用方法：" + method.getName() + "失败");

    }

    private Channel createAvailableChannel(InetSocketAddress inetSocketAddress) {
        Channel channel = CHANNEL_MAP.get(inetSocketAddress);
        if (channel == null) {
            //异步操作,获取通道,异步等待提供者发来消息
            CompletableFuture<Channel> channelFuture = new CompletableFuture<>();
            NettyBootstrapInitializer.getBootstrap().connect(inetSocketAddress).addListener(
                    (ChannelFutureListener) promise -> {
                        if (promise.isDone()) {
                            //异步获取结果
                            log.info("channel：{}建立连接成功", inetSocketAddress);
                            channelFuture.complete(promise.channel());
                        } else if (!promise.isSuccess()) {
                            log.error("channel：{}建立连接失败", inetSocketAddress);
                            channelFuture.completeExceptionally(promise.cause());
                        }
                    });
            try {
                channel = channelFuture.get(3, TimeUnit.SECONDS);
            } catch (Exception e) {
                log.error("获取通道时发生异常");
                throw new CrpcException(CrpcExceptionEnums.NETTY_ERROR);
            } finally {
                CRPC_REQUEST_THREAD_LOCAL.remove();
            }
            //如何等待
            CHANNEL_MAP.put(inetSocketAddress, channel);
        }
        if (channel == null) {
            throw new CrpcException(CrpcExceptionEnums.NETTY_ERROR);
        }
        return channel;
    }
}
